package io.kiki.sba.registry.server.shared.meta;

import com.github.rholder.retry.Retryer;
import com.github.rholder.retry.RetryerBuilder;
import com.github.rholder.retry.StopStrategies;
import com.github.rholder.retry.WaitStrategies;
import io.kiki.sba.registry.api.exchange.RequestException;
import io.kiki.sba.registry.api.exchange.message.Request;
import io.kiki.sba.registry.api.exchange.message.Response;
import io.kiki.sba.registry.common.model.GenericResponse;
import io.kiki.sba.registry.common.model.constants.ValueConstants;
import io.kiki.sba.registry.common.model.elector.Leader;
import io.kiki.sba.registry.common.model.elector.Lock;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.exception.MetaLeaderQueryException;
import io.kiki.sba.registry.remoting.http.HttpClient;
import io.kiki.sba.registry.server.shared.constant.CentralControlLeaderLearnModeEnum;
import io.kiki.sba.registry.server.shared.remoting.ClientSideExchanger;
import io.kiki.sba.registry.store.api.config.DefaultCommonConfig;
import io.kiki.sba.registry.store.api.elector.LockRepository;
import io.kiki.sba.registry.store.jdbc.elector.MetaJdbcLeaderElector;
import io.kiki.sba.registry.util.JsonUtils;
import io.kiki.sba.registry.util.StringFormatter;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;


public abstract class AbstractCentralControlLeaderExchanger extends ClientSideExchanger implements CentralControlLeaderExchanger {
    private static final Logger logger = LoggerFactory.getLogger(AbstractCentralControlLeaderExchanger.class);
    private static final String LEADER_KEY = "leader";
    private static final String EPOCH_KEY = "epoch";
    protected final Retryer<Leader> retryer = RetryerBuilder.<Leader>newBuilder().retryIfException().retryIfResult(Objects::isNull).withWaitStrategy(WaitStrategies.exponentialWait(1000, 3000, TimeUnit.MILLISECONDS)).withStopStrategy(StopStrategies.stopAfterAttempt(5)).build();

    private final Map<String, Leader> dataCenterToLeaderMap = new ConcurrentHashMap<>();
    @Autowired
    protected DefaultCommonConfig defaultCommonConfig;
    @Autowired
    private LockRepository lockRepository;
    private javax.ws.rs.client.Client client;

    protected AbstractCentralControlLeaderExchanger(String serverType) {
        super(serverType);

    }

    @PostConstruct
    public void init() {
        super.init();
        client = HttpClient.getInstance().getClient();
    }

    /**
     * send request to remote cluster meta leader
     */
    @Override
    public Response sendRequest(String dataCenter, Object requestBody) throws RequestException {
        Request request = new Request() {
            @Override
            public Object getRequestBody() {
                return requestBody;
            }

            @Override
            public Url getRequestUrl() {
                return new Url(getLeader(dataCenter).getLeader(), getServerPort());
            }
        };
        logger.info("[request] MetaNode Exchanger dataCenter={},request={},url={},callbackHandler={}", dataCenter, request.getRequestBody(), request.getRequestUrl(), request.getCallBack());

        try {
            return super.request(request);
        } catch (Throwable e) {
            // retry
            resetLeader(dataCenter);
            Url url = new Url(getLeader(dataCenter).getLeader(), getServerPort());
            logger.warn("[request] MetaNode Exchanger request send error!It will be retry once!Request url:{}", url);
            return super.request(request);
        }
    }

    /**
     * learn leader from remote resp
     */
    @Override
    public synchronized boolean learn(String dataCenter, Leader leader) {

        final Leader existLeader = this.dataCenterToLeaderMap.get(dataCenter);
        if (existLeader == null) {
            dataCenterToLeaderMap.put(dataCenter, leader);
            setServerIpSet(Collections.singleton(leader.getLeader()));
            return true;
        }
        if (leader.getEpoch() < existLeader.getEpoch()) {
            logger.warn("[setLeaderConflict]dataCenter={},exist={}/{},input={}/{}", dataCenter, existLeader.getEpoch(), existLeader.getLeader(), leader.getEpoch(), leader.getLeader());
            return false;
        } else
            if (existLeader.getEpoch() < leader.getEpoch()) {
                dataCenterToLeaderMap.put(dataCenter, leader);
                setServerIpSet(Collections.singleton(leader.getLeader()));
            }
        return true;
    }

    protected abstract CentralControlLeaderLearnModeEnum getMode();

    /**
     * reset leader from remoteMetaDomain
     */
    @Override
    public Leader resetLeader(String dataCenter) {
        Leader leader = null;
        CentralControlLeaderLearnModeEnum centralControlLeaderLearnModeEnum = getMode();
        if (centralControlLeaderLearnModeEnum == CentralControlLeaderLearnModeEnum.jdbc) {
            leader = queryLeaderFromDb();
        } else
            if (centralControlLeaderLearnModeEnum == CentralControlLeaderLearnModeEnum.loadbalancer) {
                leader = queryLeaderFromRest(dataCenter);
            }

        // connect to meta leader
        connect(new Url(leader.getLeader(), getServerPort()));
        // learn leader from resp
        learn(dataCenter, leader);
        return dataCenterToLeaderMap.get(dataCenter);
    }


    @Override
    public Leader getLeader(String dataCenter) {
        Leader leader = dataCenterToLeaderMap.get(dataCenter);
        if (leader != null) {
            return new Leader(leader.getLeader(), leader.getEpoch());
        }
        return resetLeader(dataCenter);
    }

    @Override
    public void removeLeader(String dataCenter) {
        dataCenterToLeaderMap.remove(dataCenter);
    }

    protected Leader queryLeaderFromDb() {
        try {
            return retryer.call(() -> {
                Lock lock = lockRepository.queryDistLock(MetaJdbcLeaderElector.lockName);
                if (!validateLockLeader(lock)) {
                    return null;
                }
                String leader = lock.getLockOwner();
                long epoch = lock.getGmtModifiedUnixMillis();
                return new Leader(leader, epoch);
            });
        } catch (Throwable e) {
            throw new MetaLeaderQueryException(StringFormatter.format("query meta leader error from db failed"), e);
        }
    }

    private boolean validateLockLeader(Lock lock) {
        if (lock == null) {
            logger.error("[resetLeaderFromDb] failed to query leader from db: lock null");
            return false;
        }
        long expireTimestamp = lock.getGmtModifiedUnixMillis() + lock.getDuration() / 2;
        long now = System.currentTimeMillis();
        if (expireTimestamp < now) {
            logger.error("[resetLeaderFromDb] failed to query leader from db: lock expired {}", lock);
            return false;
        }
        return true;
    }

    private Leader queryLeaderFromRest(String dataCenter) {
        Collection<String> centralControlServerDomains = getCentralControlServerDomains(dataCenter);

        try {
            return retryer.call(() -> queryLeaderInfo(dataCenter, centralControlServerDomains, client));
        } catch (Throwable e) {
            throw new MetaLeaderQueryException(StringFormatter.format("query meta leader from {} failed", centralControlServerDomains), e);
        }
    }

    private Leader queryLeaderInfo(String dataCenter, Collection<String> metaDomains, javax.ws.rs.client.Client client) {
        for (String metaDomain : metaDomains) {
            String url = String.format(ValueConstants.META_LEADER_QUERY_URL, metaDomain);
            try {
                javax.ws.rs.core.Response response = client.target(url).request().buildGet().invoke();
                Leader leader = handleResp(dataCenter, url, response);
                if (leader != null) {
                    return leader;
                }
            } catch (Throwable e) {
                logger.error("[resetLeaderFromRestServer] dataCenter:{} failed to query from url: {}", dataCenter, url, e);
            }
        }
        return null;
    }

    protected Leader handleResp(String dataCenter, String url, javax.ws.rs.core.Response response) {
        if (response.getStatus() != javax.ws.rs.core.Response.Status.OK.getStatusCode()) {
            logger.error("[resetLeaderFromRestServer] dataCenter:{} failed to query from url: {}, resp status: {}", dataCenter, url, response.getStatus());
            return null;
        }
        GenericResponse genericResponse = new GenericResponse<>();
        genericResponse = response.readEntity(genericResponse.getClass());

        if (!genericResponse.isSuccess() || genericResponse.getData() == null) {
            logger.error("[resetLeaderFromRestServer] dataCenter:{} failed to query from url: {}, resp: {}", dataCenter, url, JsonUtils.writeValueAsString(genericResponse));
            return null;
        }
        Map data = (Map) genericResponse.getData();
        Long epoch = (Long) data.get(EPOCH_KEY);
        String leader = (String) data.get(LEADER_KEY);
        if (StringUtils.isBlank(leader)) {
            return null;
        }
        Leader leader1 = new Leader(leader, epoch);
        logger.info("[resetLeaderFromRestServer] dataCenter:{} query from url: {}, meta leader:{}", dataCenter, url, leader1);
        return leader1;
    }

    protected abstract Collection<String> getCentralControlServerDomains(String dataCenter);
}
